package mq

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

//type OnReceiver interface {
//	onReceive(msg *primitive.MessageExt)
//}

type OnReceiver func(msg *primitive.MessageExt)

type Consumer struct {
	nameServer string
	topic      string
	groupName  string
	consumer   *rocketmq.PushConsumer
	_receiver  OnReceiver
}

func NewConsumer(server, topicSubs, groupName string, receiver OnReceiver) *Consumer {
	c := &Consumer{
		nameServer: server,
		topic:      topicSubs,
		groupName:  groupName,
		_receiver:  receiver,
		consumer:   nil,
	}
	c.newConsumer()
	return c
}

func (this *Consumer) Shutdown() error {
	if this.consumer != nil {
		//关闭连接
		err := (*this.consumer).Shutdown()
		if err != nil {
			fmt.Printf("shutdown consumer error: %s", err.Error())
			return err
		}
	}
	return nil
}

func (this *Consumer) newConsumer() error {
	//启动recketmq并设置负载均衡的Group
	c, err1 := rocketmq.NewPushConsumer(
		consumer.WithNameServer([]string{this.nameServer}),
		consumer.WithGroupName(this.groupName),
	)
	if err1 != nil {
		fmt.Println("NewPushConsumer失败：", err1)
		return err1
	}
	//订阅消息
	if err := c.Subscribe(this.topic, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, value := range msgs {
			//fmt.Printf("--->%v \n", value)
			if this._receiver != nil {
				this._receiver(value)
			}
		}
		return consumer.ConsumeSuccess, nil
	}); err != nil {
		fmt.Println("Subscribe failed->", err.Error())
		return err
	}
	//启动
	err := c.Start()
	if err != nil {
		fmt.Println(err.Error())
		//os.Exit(-1)
		return err
	}
	this.consumer = &c
	return nil
}
